package com.binance.zkconfig.app.watcher;

import com.binance.zkconfig.app.constant.Constant;
import com.binance.zkconfig.app.listener.ZkConfigChangeListener;
import com.binance.zkconfig.app.listener.event.ApplicationReadyEvent;
import com.binance.zkconfig.app.listener.event.ZkConfigChangeEvent;
import com.binance.zkconfig.app.service.ZkConfig;
import com.binance.zkconfig.app.util.ZkConfigUtils;
import com.google.common.base.Objects;
import com.netflix.config.ConfigurationManager;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.configuration.AbstractConfiguration;
import org.apache.commons.configuration.event.ConfigurationEvent;
import org.apache.commons.configuration.event.ConfigurationListener;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.apache.zookeeper.KeeperException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.EnvironmentAware;
import org.springframework.context.event.EventListener;
import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.core.env.Environment;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.io.Closeable;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;

import static org.apache.curator.framework.recipes.cache.TreeCacheEvent.Type.NODE_ADDED;
import static org.apache.curator.framework.recipes.cache.TreeCacheEvent.Type.NODE_REMOVED;
import static org.apache.curator.framework.recipes.cache.TreeCacheEvent.Type.NODE_UPDATED;

/**
 * Created by Fei.Huang on 2018/8/28.
 */
@Slf4j
public class ZkConfigWatcher implements Closeable, TreeCacheListener, EnvironmentAware {

    private final AtomicBoolean running = new AtomicBoolean(false);
    private final AtomicBoolean ready = new AtomicBoolean(false);

    private List<String> contexts;
    private CuratorFramework curatorFramework;
    private Map<String, TreeCache> caches;
    private ConfigurableEnvironment environment;

    @Autowired(required = false)
    private List<ZkConfigChangeListener> zkConfigChangeListeners;

    public ZkConfigWatcher(List<String> contexts, CuratorFramework curatorFramework) {
        this.contexts = contexts.stream().map(c -> c.startsWith(Constant.SLASH) ? c : Constant.SLASH + c).collect(Collectors.toList());
        this.curatorFramework = curatorFramework;
    }

    @Override
    public void setEnvironment(Environment environment) {
        synchronized (this) {
            this.environment = (ConfigurableEnvironment) environment;
        }
    }

    @EventListener
    public void handle(ApplicationReadyEvent event) {
        this.ready.compareAndSet(false, true);
    }

    @PostConstruct
    public void start() {
        if (this.running.compareAndSet(false, true)) {
            this.caches = new HashMap<>();
            for (String context : this.contexts) {
                try {
                    TreeCache cache = TreeCache.newBuilder(this.curatorFramework, context).build();
                    cache.getListenable().addListener(this);
                    cache.start();
                    this.caches.put(context, cache);
                } catch (KeeperException.NoNodeException e) {
                    log.warn("ZkConfig nodePath is not found when add listener, nodePath:{}", context);
                } catch (Exception e) {
                    log.error("ZkConfig error initializing listener for context:" + context, e);
                }
            }
        }
    }


    @Override
    public void close() {
        if (this.running.compareAndSet(true, false)) {
            for (TreeCache cache : this.caches.values()) {
                cache.close();
            }
            this.caches = null;
        }
    }

    @Override
    public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
        if (!this.ready.get()) {
            return;
        }
        log.info("ZkConfigs received changeEvent:{}", ZkConfigUtils.getEventDesc(event, true));
        TreeCacheEvent.Type eventType = event.getType();
        if (eventType == NODE_ADDED || eventType == NODE_REMOVED || eventType == NODE_UPDATED) {
            fireConfigUpdateEvent(event);
        }
    }

    protected String findContext(String path) {
        for (String context : this.contexts) {
            if (path.startsWith(context)) {
                return context;
            }
        }
        return null;
    }

    private synchronized void fireConfigUpdateEvent(TreeCacheEvent event) {
        if (event.getData() == null) {
            return;
        }
        TreeCacheEvent.Type eventType = event.getType();
        String context = this.findContext(event.getData().getPath());
        if (context == null) {
            log.warn("ZkConfig context is not found for nodePath:{}", event.getData().getPath());
            return;
        }
        String key = ZkConfigUtils.sanitizeKey(context, event.getData().getPath());
        String oldEnvValue = this.environment.getProperty(key);
        byte[] data = event.getData().getData();
        String value = data != null && data.length > 0 ? new String(data, StandardCharsets.UTF_8) : "";
        ZkConfig config = ZkConfigUtils.getConfig(context);
        if (eventType == NODE_ADDED) {
            config.put(key, value);
        } else if (eventType == NODE_UPDATED) {
            config.put(key, value);
        } else if (eventType == NODE_REMOVED) {
            config.remove(key);
        }
        notifyConfigChange(key, oldEnvValue, context, eventType);
    }

    public void notifyConfigChange(String key, String oldEnvValue, String namespance,
                                   TreeCacheEvent.Type eventType) {
        String newEnvValue = this.environment.getProperty(key);
        if (!Objects.equal(oldEnvValue, newEnvValue)) {
            AbstractConfiguration manager = ConfigurationManager.getConfigInstance();
            for (ConfigurationListener listener : manager.getConfigurationListeners()) {
                listener.configurationChanged(new ConfigurationEvent(this,
                        AbstractConfiguration.EVENT_SET_PROPERTY, key, newEnvValue, false));
            }
            notifyCustomizedListeners(
                    new ZkConfigChangeEvent(ZkConfigUtils.transformType(eventType), namespance, key, newEnvValue));
        }
    }

    private void notifyCustomizedListeners(ZkConfigChangeEvent event) {
        if (!CollectionUtils.isEmpty(zkConfigChangeListeners)) {
            for (ZkConfigChangeListener listener : zkConfigChangeListeners) {
                listener.onChange(event);
            }
        }
    }
}
